package spark.streaming.casePractice

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import spark.streaming.PropertiesUtil.JdbcUtil

import java.sql.{Connection, PreparedStatement}
import java.text.SimpleDateFormat
import java.util.Date

object DateAreaCityAdCountHandler {
  def main(args: Array[String]): Unit = {

    // Spark 配置信息 + StreamingContext
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("DateAreaCityAdCountHandler")
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

    // kafka环境变量
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux1:9092,linux2:9092,linux3:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )

    // 获取到流式数据
    val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("atguiguNew"), kafkaPara)
    )
    // 封装流式数据
    val adClickData: DStream[AdClickData] = kafkaDataDS.map(
      kafkaData => {
        val data: String = kafkaData.value()
        val datas: Array[String] = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    )

    val reduceDS: DStream[((String, String, String, String), Int)] = adClickData.map(
      data => {
        val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
        val day:String = sdf.format(new Date(data.ts.toLong))
        val area: String = data.area
        val city: String = data.city
        val ad: String = data.ad

        ((day, area, city, ad), 1)
      }
    ).reduceByKey(_ + _)

    // 对DStream中的RDD遍历处理
    reduceDS.foreachRDD(
      rdd => {
        // 每个RDD在各自分区内
        rdd.foreachPartition(
          iter => {
            val conn: Connection = JdbcUtil.getConnection
            val pstat: PreparedStatement = conn.prepareStatement(
              """
                | insert into area_city_ad_count ( dt, area, city, adid, count )
                | values ( ?, ?, ?, ?, ? )
                | on DUPLICATE KEY
                | UPDATE count = count + ?
                            """.stripMargin)
            iter.foreach {
              case ((day, area, city, ad), sum) =>
                pstat.setString(1, day)
                pstat.setString(2, area)
                pstat.setString(3, city)
                pstat.setString(4, ad)
                pstat.setInt(5, sum)
                pstat.setInt(6, sum)
                pstat.executeUpdate()
            }
            pstat.close()
            conn.close()
          }
        )
      }
    )


    ssc.start()
    ssc.awaitTermination()
  }

  // 广告点击数据
  case class AdClickData(ts: String, area: String, city: String, user: String, ad: String)
}
